/**
    向Kafka发消息模拟购买事件
 */
package com.demo.kafka.flink.shop;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

public class KafkaProducer {


    public static void main(String[] args) throws Exception{
        //首先要创建StreamExecutionEnvironment，这是流计算执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //source是程序的数据源输入，通过addSource(sourceFunction)来为程序添加一个source
        //flink提供了大量的已经实现好的source方法，可以自定义source
        //这里是通过实现sourceFunction接口来自定义一个并行度为1的source
        DataStreamSource<String> text = env.addSource(new FlinkSource()).setParallelism(1);

        //Properties类用于设置配置文件中的参数
        Properties properties = new Properties();

        //调用setProperty方法将设置的参数保存到配置文件中
        //这里是设置了kafka的bootstrap.servers参数，为启动kafka的主机ip和端口
        properties.setProperty("bootstrap.servers", "139.9.79.13:19092");

        //创建生产者producer，指定消息要发送到的topic以及添加上面的properties
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>("shop-test",new SimpleStringSchema(),properties);

        //自定义的接收器，接收producer产生的消息
        text.addSink(producer);

        //execute()方法触发程序执行
        env.execute();
    }
}